package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.event.model.broker.CreateTopicEvent;
import cn.me.alphamq.common.model.broker.CommitLogFileModel;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.common.model.broker.ConsumeQueueFileModel;
import cn.me.alphamq.broker.store.SlaveCtxManager;
import cn.me.alphamq.common.util.FileNameUtil;
import cn.me.alphamq.common.entity.req.CreateTopicReq;
import cn.me.alphamq.common.entity.resp.CreateTopicResp;
import cn.me.alphamq.common.enumeration.BrokerRoleEnum;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.AssertUtil;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 创建 topic 的事件监听器
 *
 * @author f
 */
public class CreateTopicListener implements Listener<CreateTopicEvent> {

    private static final Logger log = LoggerFactory.getLogger(CreateTopicListener.class);

    private static final String TOPIC_EXISTS = "topic 已存在";
    private static final String QUEUE_SIZE_OUT_OF_RANGE = "queueSize 应在 1~32 范围内";

    @Override
    public void onReceive(CreateTopicEvent event) throws Exception {
        CreateTopicReq createTopicReq = event.getCreateTopicReq();
        String msgId = createTopicReq.getMsgId();
        String topicName = createTopicReq.getTopicName();
        int queueSize = createTopicReq.getQueueSize();
        boolean isSyncMsg = msgId != null;

        // 如果该节点是主节点，则做参数校验
        BrokerRoleEnum role = Container.getClusterProperties().getBrokerRole();
        boolean isMaster = BrokerRoleEnum.MASTER == role;
        if (isMaster) {
            try {
                AssertUtil.isNull(Container.getCommitLogMMapFileModelManager().get(topicName), TOPIC_EXISTS);
                AssertUtil.isInClosedRange(queueSize, 1, 32, QUEUE_SIZE_OUT_OF_RANGE);
            } catch (Exception e) {
                if (isSyncMsg) {
                    event.getCtx().writeAndFlush(new TcpMsg(
                            EventCodeEnum.CREATE_TOPIC_RESP,
                            JSON.toJSONBytes(new CreateTopicResp(msgId, e.getMessage()))
                    ));
                }
                return;
            }
        }

        // 将 topic 的模型加入到 TopicModelList 中
        addTopicModelIntoTopicModelList(topicName, queueSize);
        // 创建 topic 相关的 commitLog 和 consumeQueue 文件
        createFile(topicName, queueSize);
        // 预映射 topic 相关的 commitLog 和 consumeQueue 文件
        prepareMMapLoading(topicName);

        // 打印创建 topic 的日志
        log.info("topic [{}] is created! queueSize is {}", topicName, queueSize);

        // 如果该 broker 是从节点，则无需返回结果
        if (BrokerRoleEnum.SLAVE != role) {
            if (isSyncMsg) {
                event.getCtx().writeAndFlush(new TcpMsg(
                        EventCodeEnum.CREATE_TOPIC_RESP,
                        JSON.toJSONBytes(new CreateTopicResp(msgId))
                ));
            }
        }

        // 如果该 broker 是主节点，则需要将创建 topic 的请求发送给所有从节点
        if (isMaster) {
            SlaveCtxManager.writeAndFlush(EventCodeEnum.CREATE_TOPIC_REQ, createTopicReq);
        }
    }

    // 将 topic 的模型加入到 TopicModelList 中
    private void addTopicModelIntoTopicModelList(String topicName, int queueSize) {
        String fileName = FileNameUtil.generateFileName();

        // 创建 commitLog 文件的模型
        CommitLogFileModel commitLogFileModel = new CommitLogFileModel();
        commitLogFileModel.setFileName(fileName);
        commitLogFileModel.setOffset(new AtomicInteger(0));
        commitLogFileModel.setOffsetLimit(BrokerConstant.COMMIT_LOG_MAX_SIZE);

        // 创建 consumeQueue 文件的模型
        List<ConsumeQueueFileModel> queueList = new ArrayList<>(queueSize);
        for (int queueId = 0; queueId < queueSize; queueId++) {
            ConsumeQueueFileModel consumeQueueFileModel = new ConsumeQueueFileModel();
            consumeQueueFileModel.setId(queueId);
            consumeQueueFileModel.setFileName(fileName);
            consumeQueueFileModel.setOffset(new AtomicInteger(0));
            consumeQueueFileModel.setOffsetLimit(BrokerConstant.CONSUME_QUEUE_MAX_SIZE);
            queueList.add(consumeQueueFileModel);
        }

        // 创建 topic 的模型
        long currentTimestamp = System.currentTimeMillis();
        CommitLogTopicModel commitLogTopicModel = new CommitLogTopicModel();
        commitLogTopicModel.setTopic(topicName);
        commitLogTopicModel.setCreateAt(currentTimestamp);
        commitLogTopicModel.setUpdateAt(currentTimestamp);
        commitLogTopicModel.setCommitLogFileModel(commitLogFileModel);
        commitLogTopicModel.setQueueList(queueList);

        // 将 topic 的模型加入到 TopicModelList 中
        Container.getCommitLogTopicModelList().add(commitLogTopicModel);
    }

    // 创建 topic 相关的 commitLog 和 consumeQueue 文件
    private void createFile(String topicName, int queueSize) throws IOException {
        String commitLogDirPath = FileNameUtil.getCommitLogFileDir(topicName);
        // 创建 commitLog 文件目录
        new File(commitLogDirPath).mkdir();
        // 创建 commitLog 文件
        new File(commitLogDirPath + FileNameUtil.generateFileName()).createNewFile();

        String consumeQueueBaseDirPath = FileNameUtil.getConsumeQueueFileBaseDir(topicName);
        // 创建 consumeQueue 文件父目录
        new File(consumeQueueBaseDirPath).mkdir();

        for (int queueId = 0; queueId < queueSize; queueId++) {
            String consumeQueueSubDirPath = consumeQueueBaseDirPath + "/" + queueId + "/";
            // 创建 consumeQueue 文件子目录
            new File(consumeQueueSubDirPath).mkdir();
            // 创建 consumeQueue 文件
            new File(consumeQueueSubDirPath + FileNameUtil.generateFileName()).createNewFile();
        }
    }

    // 预映射 topic 相关的 commitLog 和 consumeQueue 文件
    private void prepareMMapLoading(String topicName) throws IOException {
        Container.getCommitLogAppendHandler().prepareMMapLoading(topicName);
        Container.getConsumeQueueAppendHandler().prepareMMapLoading(topicName);
    }
}
